package com.zetyun.tiger.connector.cdc;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.MapDifference;
import com.google.common.collect.Maps;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.rt.sdk.service.cache.CachePool;
import com.zetyun.tiger.connector.util.SpecialDateFormatUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainerBuilder;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.util.TimeUtils;
import redis.clients.jedis.JedisSentinelPool;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;


import static com.zetyun.tiger.connector.cdc.CdcConstant.*;

@ActionMeta(
        id = "tiger.format-as-maxwell-data",
        tags = {"formatter"},
        category = "map",
        name = "FormatAsMaxwellData",
        parameterClass = FormatAsMaxwellDataParameter.class,
        status = ActionMeta.Status.PUBLIC,
        description = "将联盟数据格式化为类似 maxwell 采集数据的结构，" +
                "即为实时数据增加 old_data 数据字段，以及解析联盟实时数据时间字段，" +
                "格式化为人类可读的时间格式，并且转化为时间戳"
)
@Slf4j
public class FormatAsMaxwellDataAction extends MapAction {


    private SimpleRedisContainer container;
    private FormatRedisMapper redisMapper;

    private FormatAsMaxwellDataParameter parameters;
    private ObjectMapper om;

    private Integer expireTimeSeconds;

    @SneakyThrows
    @Override
    public void init(OperatorContext context) {
        super.init(context);


        om = new ObjectMapper();

        parameters = context.getParameters();

        expireTimeSeconds = parameters.getExpireTime() * 3600;
        FlinkJedisSentinelConfig jedisSentinelConfig = new FlinkJedisSentinelConfig.Builder()
                .setMasterName(parameters.getMasterName())
                .setSentinels(parameters.getSentinels())
                .build();
        GenericObjectPoolConfig genericObjectPoolConfig = RedisCommandsContainerBuilder
                .getGenericObjectPoolConfig(jedisSentinelConfig);

        redisMapper = new FormatRedisMapper(RedisCommand.HSET,
                parameters.getAdditionalKey(),
                parameters.getKeyNames(),
                om);


        JedisSentinelPool jedisSentinelPool = new JedisSentinelPool(jedisSentinelConfig.getMasterName(),
                jedisSentinelConfig.getSentinels(), genericObjectPoolConfig,
                jedisSentinelConfig.getConnectionTimeout(), jedisSentinelConfig.getSoTimeout(),
                jedisSentinelConfig.getPassword(), jedisSentinelConfig.getDatabase());
        this.container = new SimpleRedisContainer(jedisSentinelPool);

        this.container.open();
    }

    @Override
    public RtEvent apply(RtEvent in) throws Exception {

        RtEvent event = new RtEvent();

        event.setField(OP_TYPE, UPDATE_OP);

        String topic = in.getHeader("topic", String.class);
        String dbName = "";
        String tbName = "";

        if (topic != null) {
            String[] arr = topic.split(parameters.getTopicSeparator());

            dbName = arr[parameters.getDbIndex()];
            tbName = arr[parameters.getTbIndex()];
        }

        event.setField(DATABASE_NAME, dbName);
        event.setField(TABLE_NAME, tbName);

//        String eventDayId = "";
//        String processDayId = "";

        Object dataObject = in.getField(parameters.getDateField());
        if (Objects.nonNull(dataObject) && !dataObject.toString().isEmpty()) {
            Long dataTs = SpecialDateFormatUtil.getRecordTs(dataObject.toString());
            String formatDate = SpecialDateFormatUtil.toYmdHms(dataTs);
            event.setField(DATA_TS, dataTs);
            event.setField(HUMAN_DATA_TS, formatDate);

//            eventDayId = formatDate.substring(0, 10);
        } else {
            event.setField(DATA_TS, "");
            event.setField(HUMAN_DATA_TS, "");
        }

        long procTs = System.currentTimeMillis();
        String currentSystemDate = SpecialDateFormatUtil.toYmdHms(procTs);
        event.setField(PROC_TS, procTs);
        event.setField(HUMAN_PROC_TS, currentSystemDate);


//        processDayId = currentSystemDate.substring(0,10);

        event.setField(DATA, in.getFields());
        event.setField(OLD_DATA, Collections.emptyMap());

//        event.setField("add_stats", Collections.emptyMap());
//        event.setField("delete_stats", Collections.emptyMap());

        String key = redisMapper.getKeyFromData(in);
        String value = redisMapper.getValueFromData(in);

        String opType = in.getField(parameters.getOpType(), String.class);

        switch (opType) {
            case INSERT_OP:
                this.container.hset(parameters.getAdditionalKey(), key, value, null);
                event.setField(OP_TYPE, INSERT_OP);
                break;
            case UPDATE_OP:
                String lastStringVal = this.container.hget(parameters.getAdditionalKey(), key);
                if (!Objects.isNull(lastStringVal)) {
                    // 序列化先前json，比对差异
                    HashMap<String, Object> lastMap = om.readValue(lastStringVal, new TypeReference<HashMap<String, Object>>() {
                    });

                    MapDifference<String, Object> difference = Maps.difference(in.getFields(), lastMap);


                    if (difference.areEqual()) {
                        log.warn("data duplication: {}", lastStringVal);
                    } else {
                        // 键相同但是值不同值映射项。返回的Map的值类型为MapDifference.ValueDifference，
                        // 以表示左右两个不同的值
                        HashMap<String, Object> updateMap = new HashMap<>();
                        for (Map.Entry<String, MapDifference.ValueDifference<Object>> entry : difference.entriesDiffering().entrySet()) {
                            String diffFileName = entry.getKey();
                            MapDifference.ValueDifference<Object> diffValue = entry.getValue();
                            Object new_val = diffValue.leftValue();
                            Object old_val = diffValue.rightValue();

                            if (!new_val.toString().equals(old_val.toString())) {
//                                updateMap.put(diffFileName, Maps.newHashMap(
//                                        ImmutableMap.of("new_val", new_val, "old_val", old_val)));

                                updateMap.put(diffFileName, old_val);

                            }
                        }

                        event.setField(OLD_DATA, updateMap);
                    }
                    // 键只存在于左边Map的映射项
//                    event.setField("add_stats", Maps.newHashMap(difference.entriesOnlyOnLeft()));
//                    event.setField("delete_stats", Maps.newHashMap(difference.entriesOnlyOnRight()));
                }
                this.container.hset(parameters.getAdditionalKey(), key, value, null);
                break;

            case DELETE_OP:
                this.container.hdel(parameters.getAdditionalKey(), key);
                event.setField(OP_TYPE, DELETE_OP);
                break;
            default:
                throw new IllegalArgumentException("Cannot process such op type: " + opType);
        }
        return event;
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (this.container != null) {
            this.container.close();
        }
    }
}
